package com.glodon.paas.storage.ali;

import java.io.InputStream;
import java.util.ArrayList;
import java.util.List;

import com.aliyun.openservices.ClientException;
import com.aliyun.openservices.oss.OSSClient;
import com.aliyun.openservices.oss.OSSException;
import com.aliyun.openservices.oss.model.AbortMultipartUploadRequest;
import com.aliyun.openservices.oss.model.CompleteMultipartUploadRequest;
import com.aliyun.openservices.oss.model.CompleteMultipartUploadResult;
import com.aliyun.openservices.oss.model.GetObjectRequest;
import com.aliyun.openservices.oss.model.InitiateMultipartUploadRequest;
import com.aliyun.openservices.oss.model.InitiateMultipartUploadResult;
import com.aliyun.openservices.oss.model.OSSObject;
import com.aliyun.openservices.oss.model.OSSObjectSummary;
import com.aliyun.openservices.oss.model.ObjectListing;
import com.aliyun.openservices.oss.model.ObjectMetadata;
import com.aliyun.openservices.oss.model.PartETag;
import com.aliyun.openservices.oss.model.PutObjectResult;
import com.aliyun.openservices.oss.model.UploadPartRequest;
import com.aliyun.openservices.oss.model.UploadPartResult;
import com.glodon.paas.exception.PaasServerExceptionFactory;
import com.glodon.paas.storage.AbstractBulkStorage;
import com.glodon.paas.storage.InitResult;
import com.glodon.paas.storage.Part;
import com.glodon.paas.storage.ReadResult;
import com.glodon.paas.storage.WriteResult;

/**
 * An Aliyun OSS-based implementation for BulkStorage 
 * @author guojf
 *
 */
public class OSSBulkStorage extends AbstractBulkStorage {
	private static final Object LOCK = new Object();
	
    private String bucketName;
    private OSSClient client;
    
    public OSSBulkStorage(String ossEndpoint, String accessId, String accessKey, String bucketName) {
    	client = new OSSClient(ossEndpoint, accessId, accessKey);
    	this.bucketName = bucketName;
    }

    
    @Override
    public InitResult initMultiPartUpload(String key){
		try {
			InitiateMultipartUploadRequest initiatemultipartuploadrequest = new InitiateMultipartUploadRequest(bucketName,key);
			InitiateMultipartUploadResult result = client.initiateMultipartUpload(initiatemultipartuploadrequest);
			return new InitResult(result.getKey(),result.getUploadId());
		} catch (Exception er) {
			throw PaasServerExceptionFactory.storageException("failed to initiate multipart Upload object to OSS: KEY = " + key, er);
		}
    }
    @Override
    public void abortMultiPartUpload(InitResult handle){
		try {
			AbortMultipartUploadRequest abortmultipartuploadrequest = new AbortMultipartUploadRequest(bucketName,handle.getKey(),handle.getUploadId());
			client.abortMultipartUpload(abortmultipartuploadrequest);
		} catch (Exception er) {
			throw PaasServerExceptionFactory.storageException("failed to abortt Upload object to OSS: KEY = " + handle.getKey(), er);
		}
    }
    @Override
    public WriteResult completeMultiPartUpload(InitResult handle, List<Part> parts){
		try {
			List<PartETag> partETags = new ArrayList<PartETag>();
			for(Part part : parts){
				PartETag partEtag = new PartETag(part.getIndex(), part.getETag());
				partETags.add(partEtag);
			}
			CompleteMultipartUploadRequest completeMultipartUploadRequest = new CompleteMultipartUploadRequest(bucketName,handle.getKey(),handle.getUploadId(),partETags);
			CompleteMultipartUploadResult result = client.completeMultipartUpload(completeMultipartUploadRequest);
			return new WriteResult(result.getKey(), result.getETag());
		} catch (Exception er) {
			throw PaasServerExceptionFactory.storageException("failed to create OSS object group: KEY = " + handle.getKey(), er);
		}
    }
    
	@Override
	public WriteResult write(String key, InputStream object, long length) {
		try {
			ensureBucket(client, bucketName);
			ObjectMetadata objectMeta = new ObjectMetadata();
			objectMeta.setContentLength(length);
			PutObjectResult result = client.putObject(bucketName, key, object, objectMeta);
			return new WriteResult(key, result.getETag());
		} catch (Exception er) {
			throw PaasServerExceptionFactory.storageException("failed to write object to OSS: KEY = " + key, er);
		}
	}

	@Override
	public ReadResult read(String key) {
		try {
			GetObjectRequest request = new GetObjectRequest(bucketName, key);
			OSSObject result = client.getObject(request);
			return new ReadResult(result.getObjectContent(), result.getObjectMetadata().getContentLength());
		} catch (Exception er) {
			throw PaasServerExceptionFactory.storageException("failed to read object from OSS: KEY = " + key, er);
		}
	}
	
	@Override
	public boolean isPartialReadSupported() {
		return true;
	}
	
	@Override
	public boolean isMultiPartUploadSupported() {
		return true;
	}
	
	@Override
	public ReadResult read(String key, long start, long end) {
		try {
			GetObjectRequest request = new GetObjectRequest(bucketName, key);
			request.setRange(start, end);
			OSSObject result = client.getObject(request);
			ReadResult response = new ReadResult(result.getObjectContent(), result.getObjectMetadata().getContentLength());
			return response;
		} catch (Exception er) {
			throw PaasServerExceptionFactory.storageException("failed to read object from OSS: KEY = " + key, er);
		}
	}
	
	@Override
	public void delete(String key) {
		try {
            ObjectListing objects = client.listObjects(bucketName, key);
            List<OSSObjectSummary> objectSummaries = objects.getObjectSummaries();
            int size = objectSummaries.size();
            if (size == 0) {
                throw PaasServerExceptionFactory.notFoundException("not found deleting resource: " + key, null);
            } else if (size == 1) {
                client.deleteObject(bucketName, key);
            } else {
                while (size > 0) {
                    for (int i = 0; i < size; i++) {
                        String subKey = objectSummaries.get(i).getKey();
                        client.deleteObject(bucketName, subKey);
                    }
                    objects = client.listObjects(bucketName, key);
                    objectSummaries = objects.getObjectSummaries();
                    size = objectSummaries.size();
                }
            }
		} catch (Exception er) {
			throw PaasServerExceptionFactory.storageException("failed to remove object from OSS: KEY = " + key, er);
		}
	}
	
	@Override
	public WriteResult uploadPart(InitResult handle,Part part, InputStream is) {
		try {
			UploadPartRequest uploadpartrequest = new UploadPartRequest();
			uploadpartrequest.setBucketName(bucketName);
			uploadpartrequest.setKey(handle.getKey());
			uploadpartrequest.setInputStream(is);
			uploadpartrequest.setMd5Digest(part.getETag());
			uploadpartrequest.setPartNumber(part.getIndex());
			uploadpartrequest.setPartSize(part.getSize());
			uploadpartrequest.setUploadId(handle.getUploadId());
			UploadPartResult result = client.uploadPart(uploadpartrequest);
			return new WriteResult(handle.getKey(), result.getETag());
		} catch (Exception er) {
			throw PaasServerExceptionFactory.storageException("failed to create OSS object group: KEY = " + handle.getKey(), er);
		}
	}
	
	/**
	 * check if the specified bucket existing, and create it if not existing.
	 * @param client
	 * @param bucketName
	 * @throws OSSException
	 * @throws ClientException
	 */
	private void ensureBucket(OSSClient client, String bucketName) throws OSSException, ClientException{
		if (!client.isBucketExist(bucketName)){
			synchronized(LOCK) {
				if (!client.isBucketExist(bucketName)) {
					client.createBucket(bucketName);
				}
			}
		}	
	}
}
